
#include "manager.h"
#include "scoreboard.h"

void rp_rstatd_manager(int fd, short event, void *arg)
{
	char fname[] = "rp_rstatd_manager";
	char msg[] = "fatal error";
	Err err;
	int extra;
	static char fpath[MAX_FILENAME_LENGTH];	
	Worker *temp, *pworker;

	/*	re-install the timer	  */
	if (evtimer_add(&global.ev_hb, &config.tv_heartbeat) < 0)
	{
		rp_log(LOG_CRITICAL, fname, "Adding timer event failed");
	}
	
	/*	collect the host info and send the heartbeat message   */
	err = rp_rstatd_hearbeat();
	
	/*	cleanup the going-away rpointd   */
	err = rp_rstatd_scoreboard_cleanup(&global.sb_dead);
	err = rp_rstatd_scoreboard_cleanup(&global.sb_live);
	
	/*	finish the previous command   */
	switch (global.code)
	{
		case (OP_SHUTDOWN) : {
			pworker = rp_rstatd_scoreboard_remove_idle(&(global.sb_live), 0);
			while (pworker) 
			{
				temp = pworker;
				err = rp_score_set(pworker->score, NULL, OP_SHUTDOWN, NULL);
				pworker->exitFlag = 1;
				pworker = pworker->next;
				err = rp_rstatd_scoreboard_register(&(global.sb_dead), temp);
			}			
			if (global.sb_live.total == 0)
			{	/*	No running rpointd, break from the loop & we will never get back to here   */	
				err = rp_rstatd_scoreboard_cleanup(&global.sb_dead);
				struct timeval time;
				time.tv_sec = 0;
				time.tv_usec = 0;
				event_base_loopexit(global.evb_server, &time);		
				}		
			return;
		}
		case (OP_PAUSE) : {			
			return;
		}
		default : {} // TODO:
	}

	/*	manage rpointd process	    */	
	err = rp_rstatd_scoreboard_monitor(&global.sb_live, &extra);
	if (extra < 0)
	{	/*	mark the exit state & save the pworker onto ptable   */
		pworker = rp_rstatd_scoreboard_remove_idle(&(global.sb_live), abs(extra));  //peng
		while (!pworker) 
		{
			temp = pworker;
			pworker = pworker->next;
			temp->prev = NULL;
			temp->next = NULL;
			if (!pworker) pworker->prev = NULL;
			err = rp_score_set(pworker, NULL, OP_SHUTDOWN, NULL);
			err = rp_rstatd_scoreboard_register(&global.sb_dead, temp);
		}
	} else if (extra > 0) 
	{	/*	create new rpointd		*/
	        int i, fd;
		char exefile[MAX_FILENAME_LENGTH];
		
		memset(exefile, 0, MAX_FILENAME_LENGTH);
		sprintf(exefile, "%s/rpointd", env.bindir);
	        for (i=0; i<abs(extra); i++)
		{		
			pworker = (Worker*)malloc(sizeof(Worker));
			pworker->exitFlag = 0;
			
			if ((pworker->pid = rp_process_create_with_pipe(exefile, NULL, environ, global.fd_pod)) <= 0)
			{
				free(pworker);
				rp_log(LOG_CRITICAL, fname, "Fail to create new rpointd process");
				continue;
			} 
		
			sprintf(fpath, "/tmp/rally/%d", pworker->pid);
			if ((fd = open(fpath, O_RDWR|O_CREAT|O_TRUNC, 00777)) < 0)
			{
				err = rp_process_terminate(pworker->pid, SIGTERM);		// Forceful kill the new rpointd process
				pworker->exitFlag = 1;
				err = rp_score_init_pshared(pworker->score);
				err = rp_rstatd_scoreboard_register(&global.sb_dead, pworker);
				rp_log(LOG_CRITICAL, fname, "Cannot open file under /tmp/rally");
				continue;
			} 
			lseek(fd, sizeof(Score), SEEK_END);
			write(fd, "", 1);		// mmap size = Score + 1
			if ((pworker->score = mmap(0, sizeof(Score), PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0)) == MAP_FAILED)
			{
				close(fd);
				err = rp_process_terminate(pworker->pid, SIGTERM);		// Forceful kill the new rpointd process
				pworker->exitFlag = 1;
				err = rp_score_init_pshared(pworker->score);
				err = rp_rstatd_scoreboard_register(&global.sb_dead, pworker);
				rp_log(LOG_CRITICAL, fname, "Failed mmap file under /tmp/rally");
				continue;
			} 
			close(fd);
			
			err = rp_score_init_pshared(pworker->score);
			err = rp_rstatd_scoreboard_register(&global.sb_live, pworker);
		}
	}
	
	return;
}


